package com.syp.test.clustercacheflush.util.zkflushcache.handler;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkConnection;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.syp.test.clustercacheflush.util.Constants;
import com.syp.test.clustercacheflush.util.zkflushcache.ZkCacheProperties;
import com.syp.test.clustercacheflush.util.zkflushcache.ZkCacheSerializer;

@Component
public class CacheSyncProcesser implements InitializingBean, DisposableBean {

	@Autowired
	private ZkCacheProperties properties;
	@Autowired
	private ZkCacheSerializer zkCacheSerializer;
	@Autowired
	private CacheSyncListener cacheSyncListener;

	private ZkClient zk;

	@Override
	public void afterPropertiesSet() throws Exception {
		ZkConnection zkconn = null;
		if (properties.getSessionTimeout() > 0) {
			zkconn = new ZkConnection(properties.getZkServers(), properties.getSessionTimeout());
		} else {
			zkconn = new ZkConnection(properties.getZkServers());
		}

		if (properties.getConnectionTimeout() > 0 && properties.getOperationRetryTimeout() > 0) {
			zk = new ZkClient(zkconn, properties.getConnectionTimeout(), zkCacheSerializer, properties.getOperationRetryTimeout());
		} else if (properties.getConnectionTimeout() > 0) {
			zk = new ZkClient(zkconn, properties.getConnectionTimeout());
		} else {
			zk = new ZkClient(zkconn);
		}
		zk.subscribeChildChanges(Constants.CLUSTER_CACHE_FLUSH_ZK_PATH, cacheSyncListener);
	}

	@Override
	public void destroy() throws Exception {
		zk.close();
	}

}